package netty.tcp;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.CharsetUtil;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Demo01 {

    static ExecutorService executorService = Executors.newFixedThreadPool(10);

    public static void main(String[] args) throws InterruptedException {

        Executors.newSingleThreadExecutor().execute(Demo01::server);

        Thread.sleep(3000);

        executorService.execute(Demo01::client);

//        for (int i = 0; i < 1000; i++) {
//            Thread.sleep(500);
//            executorService.execute(Demo01::client);
//        }

        Thread.currentThread().join();
    }

    public static void server() {

        // 默认情况下时CPU线程数 * 2
        // 如果只监听一个端口，bossGroup似乎只会启动一个线程
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(
                4,
                new ThreadFactoryBuilder().setNameFormat("ServerBossThread-%d")
                        .setDaemon(true)
                        .build()
        );
        // 可以自定义线程数
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(
                32,
                new ThreadFactoryBuilder().setNameFormat("ServerWorkThread-%d")
                        .setDaemon(true)
                        .build()
        );

        ServerBootstrap serverBootstrap = new ServerBootstrap();

        try {
            serverBootstrap.group(bossGroup, workerGroup) // 设置两个线程组
                    .channel(NioServerSocketChannel.class) // 使用 NioServerSocketChannel 作为服务端的通道实现
                    .option(ChannelOption.SO_BACKLOG, 128) // 设置线程队列进程个数
                    .childOption(ChannelOption.SO_KEEPALIVE, true) // 设置保持活动连接状态
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        // 可以在这里给 WorkGroup 的 pipeline 设置一系列处理器
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new ServerHandle());
                        }
                    });

            System.out.println("服务器配置完毕");

            // 绑定端口并启动服务端
            ChannelFuture channelFuture = serverBootstrap.bind(7788).sync();

            System.out.println("服务器启动完成");

            // 注册监听器，监听关注的事件
            channelFuture.addListener(future -> {
                if (future.isSuccess()) {
                    System.out.println("监听到绑定成功~");
                } else {
                    System.out.println("监听到绑定失败！");
                }
            });

            // 对关闭通道进行监听
            // 注意：这里是阻塞的，如果不阻塞下面就不能写关闭的代码，否则会直接关闭服
            channelFuture.channel().closeFuture().sync();
            System.out.println("==Server OVER==");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 优雅关闭
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    // 测试一下多个客户端能不能复用一个 NioEventLoopGroup
    static NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(
            4,
            new ThreadFactoryBuilder().setNameFormat("ClientEventLoopGroup-%d")
                    .setDaemon(true)
                    .build()
    );

    public static void client() {

        Bootstrap bootstrap = new Bootstrap();

        try {
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new ClientHandle());
                        }
                    });

            System.out.println("客户端配置完毕");

            // 与服务器建立连接
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 7788).sync();

            System.out.println("客户端启动完毕");

            // 设置关闭通道监听
            // 注意：这里是阻塞的，如果不阻塞下面就不能写关闭的代码，否则会直接关闭服务
            // channelFuture.channel().closeFuture().sync();
            System.out.println("==Client OVER==");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 优雅关闭（为了测试多客户端复用一个NioEventLoopGroup，就不关闭了）
            // eventLoopGroup.shutdownGracefully();
        }
    }
}

class ServerHandle extends ChannelInboundHandlerAdapter {

    /**
     * 在这里可以读取客户端发送的消息
     *
     * @param ctx 上下文对象
     * @param msg 客户端发送的数据
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        System.out.println("服务端使用的线程是：" + Thread.currentThread().getName());
        // 任务队列之：用户程序定义普通任务
        // 会将任务添加到NioEventLoop的Queue<Runnable> tailTasks中
        ctx.channel().eventLoop().execute(() -> {
            // 可以在这里异步执行一些用户任务
            System.out.println("用户程序定义普通任务使用的线程是：" + Thread.currentThread().getName());
        });
        System.out.println("非阻塞。。。");

        // 任务队列之：用户自定义定时任务
        // 会将任务添加到ScheduleTaskQueue中
        ctx.channel().eventLoop().schedule(
                () -> {
                    System.out.println("用户自定义定时任务使用的线程是：" + Thread.currentThread().getName());
                },
                10,
                TimeUnit.SECONDS
        );

        // 模拟当前线程占用中，检查刚提交的任务是由哪个线程执行（还是当前线程，因为都是同一个NioEventLoop里面的线程）
        Thread.sleep(10000);

        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("客户端发送消息：" + ctx.channel().remoteAddress() + " - " + byteBuf.toString(CharsetUtil.UTF_8));
    }

    // 数据读取完毕在这里回复消息
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello Client", CharsetUtil.UTF_8)).addListener(ChannelFutureListener.CLOSE);
    }
}

class ClientHandle extends ChannelInboundHandlerAdapter {

    /**
     * 通道就绪时就会触发，在这里发消息
     *
     * @param ctx 上下文
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello Server", CharsetUtil.UTF_8));
    }

    // 同服务端
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("服务端回复消息：" + ctx.channel().remoteAddress() + " - " + byteBuf.toString(CharsetUtil.UTF_8));
    }
}
